package com.gugee.ins.data.collector.blogger.service.impl;

import com.gugee.common.exception.BizException;
import com.gugee.common.exception.GugeeException;
import com.gugee.common.exception.SpiderRetryableException;
import com.gugee.common.thread.ScheduleTemplate;
import com.gugee.ins.common.constant.InsBloggerMsgTypeCodeEnum;
import com.gugee.ins.common.dal.service.blogger.InsBloggerDao;
import com.gugee.ins.common.dal.service.spider.SpiderSvc;
import com.gugee.ins.common.model.mysql.blogger.InsBlogger;
import com.gugee.ins.common.model.spider.blogger.FollowersInfo;
import com.gugee.ins.common.model.spider.blogger.FollowingInfo;
import com.gugee.ins.data.collector.blogger.config.CollectorBloggerConfig;
import com.gugee.ins.data.collector.blogger.service.InsFeedSvc;
import com.gugee.ins.data.common.consumer.AbstractGugeeMessageHandler;
import com.gugee.ins.data.common.msg.InsBloggerMessage;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
import org.springframework.web.client.ResourceAccessException;

import javax.annotation.PostConstruct;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * Created with IntelliJ IDEA.
 * Description:
 * User: luolifeng
 * Date: 2021-01-07 14:16
 */
@Slf4j
@Service
public class InsFeedSvcImpl implements InsFeedSvc  {
    @Autowired
    ScheduleTemplate scheduleTemplate;
    /**
     * 配置
     */

    FollowerFeedInsBloggerHandler followerFeedInsBloggerHandler;
    FollowingFeedInsBloggerHandler followingFeedInsBloggerHandler;
    @Autowired
    CollectorBloggerConfig collectorBloggerConfig;
    @Autowired
    SpiderSvc spiderSvc;
    @Autowired
    KafkaTemplate<Long, InsBloggerMessage> kafkaTemplateInsBlogger;
    @Autowired
    InsBloggerDao insBloggerDao;
    @PostConstruct
    public void init() {
        //todo 线程池对象需统一生成，统一分配
        ThreadPoolExecutor customPoolMain = scheduleTemplate.getCustomThreadPool("blg-feed", collectorBloggerConfig.getInsBloggerFeedCorePollSize(),collectorBloggerConfig.getInsBloggerFeedCorePollSize(), 0, 100000, TimeUnit.MINUTES);
        followerFeedInsBloggerHandler=new FollowerFeedInsBloggerHandler(customPoolMain);
        followingFeedInsBloggerHandler=new FollowingFeedInsBloggerHandler(customPoolMain);
    }
    @Override
    public void followerFeedBloggerFromMsg(List<InsBloggerMessage> messages) {
        followerFeedInsBloggerHandler.handle(messages);
    }
    @Override
    public void followingFeedBloggerFromMsg(List<InsBloggerMessage> messages) {
        followingFeedInsBloggerHandler.handle(messages);
    }
    /**
     * 继承AbstractGugeeMessageHandler，使得具备消息处理能力
     */
    class FollowingFeedInsBloggerHandler  extends AbstractGugeeMessageHandler<InsBloggerMessage>{

        @Override
        public void handleMsg(InsBloggerMessage message) {
            FollowingInfo followingInfo= spiderFollowing(message.getUid());
            List<Long> ids=new ArrayList<>(followingInfo.getFollowers().size());

            followingInfo.getFollowers().forEach(
                    follower ->{
                        if(Boolean.FALSE.equals(follower.getIsPrivate())){
                            ids.add(follower.getUid());
                        }
                    }
            );
            List<InsBlogger> bloggers=insBloggerDao.findExistUidByUids(ids);
            Map<Long, InsBlogger> existsBloggersMap = bloggers.stream().collect(Collectors.toMap(InsBlogger::getUid, a -> a,(k1, k2)->k1));
            followingInfo.getFollowers().forEach(
                    follower ->{
                        if(!existsBloggersMap.containsKey(follower.getUid()) && Boolean.FALSE.equals(follower.getIsPrivate()) ){
                            InsBloggerMessage msg =new InsBloggerMessage();
                            msg.setUid(follower.getUid());
                            msg.setPicUrl(follower.getPicUrl());
                            msg.setUserName(follower.getUsername());
                            msg.setRanking(Boolean.FALSE);
                            kafkaTemplateInsBlogger.send(collectorBloggerConfig.getInsBloggerCollectOnlineTopic(),msg);
                        }
                    }
            );
        }
        private FollowingInfo spiderFollowing(long uid){
            return spiderSvc.spiderFollowing(uid,collectorBloggerConfig.getInsFeedFollowerTotalCount(),collectorBloggerConfig.getInsFeedFollowerBatchSize());
        }
        public FollowingFeedInsBloggerHandler(ThreadPoolExecutor customPoolMain) {
            super(customPoolMain);
        }

        @Override
        public void failHandle(Exception e,InsBloggerMessage msg) {

            if(e instanceof ResourceAccessException ){
                log.error("超时-----需重试{}次：{}" ,msg.getRetryTimes()-1,e.getMessage());
                if(msg.getRetryTimes()>0){
                    msg.setRetryTimes(msg.getRetryTimes()-1);
                    kafkaTemplateInsBlogger.send(collectorBloggerConfig.getInsBloggerCollectOnlineTopic(),msg);
                }
                if(msg.getRetryTimes()==0){
                    log.error("重试{}次后依旧失败:msg={}" ,msg.getRetryLimitTimes(),msg);
                }
            }else  if(e instanceof BizException || e instanceof  SpiderRetryableException ){
                GugeeException  bizExc=(GugeeException)e;

                if(msg.getRetryTimes()>0){
                    log.error("{}; {};需重试{}次" ,bizExc.getCode(),e.getMessage(),msg.getRetryTimes()-1);
                    msg.setRetryTimes(msg.getRetryTimes()-1);
                    kafkaTemplateInsBlogger.send(collectorBloggerConfig.getInsBloggerCollectOnlineTopic(),msg);
                }
                if(msg.getRetryTimes()==0){
                    log.error(" {};重试{}次后依旧失败:msg={}" ,bizExc.getCode(),msg.getRetryLimitTimes(),msg);
                }
            }else{
                log.error(" {}:{}" ,e.getMessage(),msg);
            }

        }

        @Override
        public void successHandle(InsBloggerMessage msg) {
            log.debug("重试{}次后成功:msg={}" ,msg.getRetryLimitTimes(),msg);
        }

    }
     class FollowerFeedInsBloggerHandler  extends AbstractGugeeMessageHandler<InsBloggerMessage>{

        @Override
        public void handleMsg(InsBloggerMessage message) {
            FollowersInfo followersInfo= spiderFollowers(message.getUid());
            List<Long> ids=new ArrayList<>(followersInfo.getFollowers().size());

            followersInfo.getFollowers().forEach(
                    follower ->{
                        if(!follower.getIsPrivate()){
                            ids.add(follower.getUid());
                        }
                    }
            );
            List<InsBlogger> bloggers=insBloggerDao.findExistUidByUids(ids);
            Map<Long, InsBlogger> existsBloggersMap = bloggers.stream().collect(Collectors.toMap(InsBlogger::getUid, a -> a,(k1, k2)->k1));
            followersInfo.getFollowers().forEach(
                    follower ->{
                        if(!existsBloggersMap.containsKey(follower.getUid()) && Boolean.FALSE.equals(follower.getIsPrivate()) ){
                            InsBloggerMessage msg =new InsBloggerMessage();
                            msg.setUid(follower.getUid());
                            msg.setPicUrl(follower.getPicUrl());
                            msg.setUserName(follower.getUsername());
                            msg.setTypeCode(InsBloggerMsgTypeCodeEnum.UID_UPDATE_F_FEED.getCode());
                            msg.setRanking(Boolean.FALSE);
                            kafkaTemplateInsBlogger.send(collectorBloggerConfig.getInsBloggerCollectOnlineTopic(),msg);
                        }
                    }
            );
        }
        private FollowersInfo spiderFollowers(long uid){
            return spiderSvc.spiderFollowers(uid,collectorBloggerConfig.getInsFeedFollowerTotalCount(),collectorBloggerConfig.getInsFeedFollowerBatchSize());
        }
        public FollowerFeedInsBloggerHandler(ThreadPoolExecutor customPoolMain) {
            super(customPoolMain);
        }

        @Override
        public void failHandle(Exception e,InsBloggerMessage msg) {

            if(e instanceof ResourceAccessException ){
                log.error("超时-----需重试{}次：{}" ,msg.getRetryTimes()-1,e.getMessage());
                if(msg.getRetryTimes()>0){
                    msg.setRetryTimes(msg.getRetryTimes()-1);
                    kafkaTemplateInsBlogger.send(collectorBloggerConfig.getInsBloggerCollectOnlineTopic(),msg);
                }
                if(msg.getRetryTimes()==0){
                    log.error("重试{}次后依旧失败:msg={}" ,msg.getRetryLimitTimes(),msg);
                }
            }else  if(e instanceof BizException || e instanceof  SpiderRetryableException ){
                GugeeException  bizExc=(GugeeException)e;

                if(msg.getRetryTimes()>0){
                    log.error("{}; {};需重试{}次" ,bizExc.getCode(),e.getMessage(),msg.getRetryTimes()-1);
                    msg.setRetryTimes(msg.getRetryTimes()-1);
                    kafkaTemplateInsBlogger.send(collectorBloggerConfig.getInsBloggerCollectOnlineTopic(),msg);
                }
                if(msg.getRetryTimes()==0){
                    log.error(" {};重试{}次后依旧失败:msg={}" ,bizExc.getCode(),msg.getRetryLimitTimes(),msg);
                }
            }else{
                log.error(" {}:{}" ,e.getMessage(),msg);
            }

        }

        @Override
        public void successHandle(InsBloggerMessage msg) {
            log.debug("重试{}次后成功:msg={}" ,msg.getRetryLimitTimes(),msg);
        }

    }
}
